package com.bclz.parallelism

import java.util.concurrent.atomic.{AtomicInteger, AtomicReference}
import java.util.concurrent.{Callable, ExecutorService}

import scala.annotation.tailrec

/**
  * 多个线程访问保证每次只有一个线程在执行处理逻辑;
  * 默认错误处理抛出异常
  * Actor模型:
  *   1.确保每个Actors在同一时间处理最多一个消息，单线程地依次执行发送给它的消息。
  *   2.不同的Actor可以同时执行它们的消息
  *   3.每个Actor之间通过发消息通信
  */
final class Actor[A](strategy:Strategy)(handle:A=>Unit,onError:Throwable=>Unit= throw(_) ) {

    //定义处理链路类
    private class HandleNode[A](var a:A=null.asInstanceOf[A]) extends AtomicReference[HandleNode[A]]
    //initial value={a=null,value=null}
    private val tail=new AtomicReference(new HandleNode[A]())
    //initial value=tail.value
    private val head=new AtomicReference(tail.get())

    private val monitorValue=new AtomicInteger(1)

    def !(a:A): Unit ={
      //{a=a,value=null}
      val handleNode = new HandleNode(a)
      //t也等于tail.get,值相当于一个链路t=a,node1->a1_1,node1_1->a1_1_1,node1_1_1...
      //getAndSet，set新值，返回老值,返回AtomicReference的value,即node1
      //lazySet，set的是node1 AtomicReference下面的value,lazySet()没有线程可见性，只是普通的set(性能比set好，不过也差不了多少，使用场景可能是不存在并发场景下的set)
      //给Node链路添节点
      //eg: 创建Actor ，tail.value= null,null  ,head=tail.value
      //第一次处理a,假定a=1, head= value{1,null}    tail= value{null.{1,null}}(getAndSet实际上是tail.value),假定handle是耗时处理
      //第二次，假定a=2, head={2,null}    tail=value{null.{1,{2,null}}}  null->1->2
      //第三次,a=3,head={3,null}   tail=value{null.{1,{2,{3,null}}}}     null->1->2->3
      //每次head set后，老值就是tail的尾节点引用(getAndSet是具有原子性,用lazySet直接set比较好)
      head.getAndSet(handleNode).lazySet(handleNode)
      trySchedule()
    }


    private def trySchedule(): Unit ={

      if(monitorValue.compareAndSet(1,0)) schedule()
    }

    private def schedule():Unit={
      strategy(act())
    }

    private def act()={

      val t = tail.get
      //分批处理
      val handleAfterNode = batchHandle(t,1024)
      if(t ne handleAfterNode){ //不等于，说明处理了1024个消息后node中还有节点没有处理完

        //handleAfterNode已经处理过a后返回
        handleAfterNode.a=null.asInstanceOf[A]
        tail.lazySet(handleAfterNode)   //一个actor同一时间单线程，用lazySet就可以了
        schedule()    //继续处理剩下的节点
      }else{
        //处理完，放弃锁
        monitorValue.set(1)

        //判断tail的子节点是否还有节点未处理
        //若有新消息进入，继续处理
        if(handleAfterNode.get ne null) trySchedule()
      }

    }

  @tailrec
  private def batchHandle(handleNode: HandleNode[A],handCount:Int): HandleNode[A] ={
    val node = handleNode.get
    if(node ne null){
      try {
        handle(node.a)
      }catch {
        case e:Exception => onError(e)
      }
      if(handCount>0) batchHandle(node,handCount-1) else node
    }else handleNode
  }


}


object Actor{


  def apply[A]( es: ExecutorService)(handle: A => Unit,onError:Throwable=>Unit=throw(_)): Actor[A] =
    new Actor[A](Strategy.strategyInExecutorService(es))(handle,onError)

}

trait Strategy{

  def apply[A](a: =>A):()=>A

}

object Strategy {

  /** 提交线程池计算策略，异步 */
  def strategyInExecutorService[A](es: ExecutorService): Strategy =
    new Strategy {
      override def apply[A](a: => A): () => A = {
        val futureA = es.submit(new Callable[A] {
          override def call(): A = a
        })
        () => futureA.get
      }
    }

  /** 单线程正常执行 */
  def strategyInSingle[A](): Strategy =
    new Strategy {
      override def apply[A](a: => A): () => A = {
        val c = a
        () => c
      }
    }

}